/*
 * This Source Code Form is subject to the terms of the Mozilla Public
 * License, v. 2.0. If a copy of the MPL was not distributed with this file,
 * You can obtain one at http://mozilla.org/MPL/2.0/.
 *
 * Portions created by the Initial Developer are
 * Copyright (C) 2012 the Initial Developer.
 * All Rights Reserved.
 *
 * The Original Code is "http://semi-frame.googlecode.com/svn/trenk/".
 *
 * The Initial Developer of the Original Code is the author below.
 */
/**
 * @file MessageQue.cpp
 * @brief auto generated.
 * @author mornongcloud
 * @date 2012/03/11
 */

#include "semi/ipc/MessageQue.h"

namespace semi {

namespace ipc {

MessageQue::MessageQue() {
}

MessageQue::MessageQue(std::string taskid) {
	this->init(taskid);
}
MessageQue::~MessageQue() {
	//	wait(0);
	if (0 < this->message)
		free(this->message);
}

void MessageQue::init(std::string taskid) {

	this->message = 0;
	//	this->mcrnt.init(taskid);
	this->TaskId = taskid;
	this->FifoPath = this->createQuename(taskid);

	if (-1 == mkfifo(this->FifoPath.c_str(), 0666)) {
		//		perror("MessageQue::init mkfifo");
	}
}

std::string MessageQue::createQuename(std::string taskid) {

	std::string name;
	name = semi::util::StringUtil::replace(taskid, "/", "");

	int siz = 32;
	if (siz < name.size())
		name = name.substr(0, siz);
	return posix::SharedMemory::getSharedMemoryPath() + "semi.que." + name;
}

std::string MessageQue::getQuePath() {
	return this->FifoPath;
}

int MessageQue::send(std::string message) {

	//	Param param;
	//	memset(&param, 0, sizeof(Param));
	//	param.message = (char*) message.c_str();
	//	param.size = message.length();
	//	//	strncpy(param.task_id, this->)
	return this->send((char*) message.c_str(), message.length());
}

int MessageQue::send(char messsage[], int size) {

	//	if (-1 == mkfifo(this->FifoPath.c_str(), 0666)) {
	//		perror("MessageQue::send mkfifo");
	//	}

	int fd;
	if (-1 == (fd = open(this->FifoPath.c_str(), O_WRONLY))) {
		//	if (-1 == (fd = open(this->FifoPath.c_str(), O_RDWR))) {
		perror("MessageQue::send open");
		return -1;
	}

	long len = size;
	char msg[len];
	memcpy(msg, messsage, len);

	int pnt = 0;
	while (pnt < len) {

		int blen = BUFFER_SIZE;
		if (len < (pnt + blen))
			blen = len - pnt;

		//		std::cout << "write(" << pnt << "-" << blen << "):" << std::string(
		//				&msg[pnt], blen) << std::endl;
		//		std::cout << util::LogUtil::HexString(&msg[pnt], blen) << std::endl;
		if (0 == pnt)
			write(fd, &len, sizeof(long));

		write(fd, &msg[pnt], blen);
		pnt = pnt + blen;
	}

	close(fd);

	return len;
}

int MessageQue::receive() {

	//	Param ret;
	//	ret.size = 0;
	//	strncpy(ret.task_id, this->TaskId.c_str(), this->TaskId.length());

	int fd = 0;
	if (-1 == (fd = open(this->FifoPath.c_str(), O_RDONLY))) {
		//	if (-1 == (fd = open(this->FifoPath.c_str(), O_RDWR))) {
		perror("MessageQue::receive open");
		//		ret.success = false;
		return 0;
	}

	long len = 0;
	read(fd, &len, sizeof(long));
	//	std::cout << "read lengh:" << len << std::endl;
	if (0 == len)
		return len;

	if (0 < this->message)
		free(this->message);
	this->message = (void*) malloc(len);
	memset((char*) this->message, 0, len);

	int pnt = 0;
	int tmp = 1;
	while (0 < tmp) {

		int blen = BUFFER_SIZE;
		if (pnt < len && len < (pnt + blen))
			blen = len - pnt;
		char buf[blen];
		memset(buf, 0, blen);

		tmp = read(fd, buf, blen);
		if (0 < tmp)
			memcpy(((char*) this->message) + pnt, buf, tmp);
		else
			pnt++;

		//		std::cout << "memcpy(" << tmp << "):" << this->message;
		//		std::cout << util::LogUtil::HexString(this->message, len) << std::endl;

		pnt = pnt + tmp;
	}
	//	memcpy(ret, msg, len);
	close(fd);

	return len;
}

void MessageQue::sweep() {
	int fd;
	if (-1 == (fd = open(this->FifoPath.c_str(), O_RDWR))) {
		perror("MessageQue::sweep open");
		return;
	}
	long len = 1;
	char buf[1];
	buf[0] = 0;

	write(fd, &len, sizeof(long));
	write(fd, buf, 1);

	close(fd);
}

void MessageQue::terminate() {

	this->sweep();

	std::stringstream ss;
	ss << "rm -f ";
	ss << this->FifoPath.c_str();
	system(ss.str().c_str());
}

}
}
